package com.wtwd.device.rocketMq;


import com.wtwd.device.mqtentity.MessageBase;
import com.wtwd.device.properties.MqProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;


@Slf4j
public class MqSendMessage {

    private DefaultMQProducer producer;

    private MqProperties mqProperties;

    public MqSendMessage(MqProperties properties) {
        this.mqProperties = properties;
        producer = new DefaultMQProducer(mqProperties.getDeviceMsgProduceGroup());
        producer.setNamesrvAddr(mqProperties.getNamesrvAddr());
//        producer.setVipChannelEnabled(false);
        producer.setSendMsgTimeout(5000);
         producer.setRetryTimesWhenSendFailed(3);
        try {
            producer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void push(String topic, String tag, MessageBase messageBase) {
        Message msg = null;
        try {
            byte[] message = messageBase.serialize();
            msg = new Message(topic,
                    tag,
                    message
            );
            SendResult sendResult = producer.send(msg, 10000);
            log.info("mq topic:{},tag:{} 推送处理服务器成功{},device:{},cmd:{}",topic,tag,sendResult.toString(), messageBase.getDeviceKey(), messageBase.toString());
        } catch (Exception e) {
            log.info("mq topic:{},tag:{} 推送处理服务器失败,device:{},cmd:{},错误:{}", topic,tag,messageBase.getDeviceKey(), messageBase.toString(),e.getMessage());
        }
    }

    public void push(MessageBase messageBase) {
        push(mqProperties.getDeviceMsgProduceTopic(), mqProperties.getDeviceMsgTag(), messageBase);
    }

}
